package com.mall.promo.mq;

import com.alibaba.fastjson.JSON;
import com.mall.order.dto.CreateSeckillOrderRequest;
import com.mall.promo.cache.CacheManager;
import com.mall.promo.dal.persistence.PromoItemMapper;
import io.netty.util.Timeout;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.LocalTransactionState;
import org.apache.rocketmq.client.producer.TransactionListener;
import org.apache.rocketmq.client.producer.TransactionMQProducer;
import org.apache.rocketmq.client.producer.TransactionSendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;
import org.redisson.Redisson;
import org.redisson.api.RLock;
import org.redisson.api.RedissonClient;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.cache.CacheProperties;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import java.nio.charset.StandardCharsets;
import java.util.HashMap;
import java.util.Map;

/**
 * 发送事务消息
 * @author 晓帆
 * @version 1.0
 **/
@Slf4j
@Component
public class PromoOrderProducer {
    @Autowired
    PromoItemMapper promoItemMapper;
    //将事务结果存入redis
    @Autowired
    CacheManager cacheManager;
    @Autowired
    RedissonClient redissonClient;

    TransactionMQProducer producer;
    @PostConstruct
    public void init() throws MQClientException {
        producer=new TransactionMQProducer("order_promotion_producer");
        producer.setNamesrvAddr("127.0.0.1:9876");

        //设置监听器
        producer.setTransactionListener(new TransactionListener() {
            @Override
            public LocalTransactionState executeLocalTransaction(Message message, Object o) {
                Map<String,Long> paramMap = (Map<String, Long>) o;
                Long productId = paramMap.get("productId");
                Long psid = paramMap.get("psid");

                //存入redis的事务执行状态的key
                String transactionKey = "promo_order_local_transaction"+message.getTransactionId();

                //使用基于redis的分布式锁
                String lockKey = "promo_item_stock_lock"+psid+"_"+productId;
                //获取锁
                RLock redissonClientLock = redissonClient.getLock(lockKey);

                //上锁
                redissonClientLock.lock();
                try {
                    Integer State = promoItemMapper.decreaseStock(productId, psid);
                    //扣除失败
                    //库存售罄，用redis打上标记

                    if (State<1){
                        String stockIsNullKey = "stock_is_null"+psid+productId;
                        cacheManager.setCache(stockIsNullKey,"stockIsNullKey",1);
                        cacheManager.setCache(transactionKey,"fail",1);
                        return LocalTransactionState.ROLLBACK_MESSAGE;
                    }
                }finally {
                    //解锁，在finally中保证一定执行
                    redissonClientLock.unlock();
                }

                cacheManager.setCache(transactionKey,"success",1);
                return LocalTransactionState.COMMIT_MESSAGE;
            }

            @Override
            public LocalTransactionState checkLocalTransaction(MessageExt messageExt) {
                String transactionKey = "promo_order_local_transaction"+messageExt.getTransactionId();
                String s = cacheManager.checkCache(transactionKey);
                if (s!=null&&"success".equals(s.trim())){
                    return LocalTransactionState.COMMIT_MESSAGE;
                }
                else if (s!=null&&"fail".equals(s.trim())){
                    return LocalTransactionState.ROLLBACK_MESSAGE;
                }

                return LocalTransactionState.UNKNOW;
            }
        });
        producer.start();
    }
    //发送消息方法
    public boolean senCreatePromoOrder(CreateSeckillOrderRequest request,Long psid,Long productId){
        String s = JSON.toJSONString(request);
        Message message = new Message("create_promotion_order",s.getBytes(StandardCharsets.UTF_8));
        HashMap<String, Long> promoMap = new HashMap<>();
        promoMap.put("productId",productId);
        promoMap.put("psid",psid);
        //参数只能传一个，所以封装进map
        TransactionSendResult sendResult;
        try {
            sendResult = producer.sendMessageInTransaction(message, promoMap);
            if (sendResult!=null&&LocalTransactionState.COMMIT_MESSAGE.equals(sendResult.getLocalTransactionState())){
                log.info("消息发送成功，本地事务执行成功");
                return true;
            }
            else if (sendResult!=null&&LocalTransactionState.ROLLBACK_MESSAGE.equals(sendResult.getLocalTransactionState())){
                log.info("消息发送成功，本地事务执行失败");
                return false;
            }
        } catch (MQClientException e) {
            e.printStackTrace();
        }
        return false;
    }
}
